package com.sdses.flink.sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class FlinkSQL_TableJoinTest {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        tableEnv.executeSql("CREATE TABLE students_table (\n" +
                "    id INT,\n" +
                "    name STRING,\n" +
                "    age INT,\n" +
                "    PRIMARY KEY (id) NOT ENFORCED\n" +
                ") WITH (\n" +
                "    'connector' = 'mysql-cdc',\n" +
                "    'hostname' = '192.168.102.155',\n" +
                "    'port' = '33066',\n" +
                "    'username' = 'datacube_test',\n" +
                "    'password' = 'Gbd#XERA@19d1ZtD',\n" +
                "    'database-name' = 'db_datacube_test',\n" +
                "    'table-name' = 'students'\n" +
                ")");

        tableEnv.executeSql("CREATE TABLE scores_table (\n" +
                "    student_id INT,\n" +
                "    math_score INT,\n" +
                "    english_score INT,\n" +
                "    PRIMARY KEY (student_id) NOT ENFORCED\n" +
                ") WITH (\n" +
                "    'connector' = 'mysql-cdc',\n" +
                "    'hostname' = '192.168.102.155',\n" +
                "    'port' = '33066',\n" +
                "    'username' = 'datacube_test',\n" +
                "    'password' = 'Gbd#XERA@19d1ZtD',\n" +
                "    'database-name' = 'db_datacube_test',\n" +
                "   'scan.incremental.snapshot.chunk.key-column' = 'student_id',\n" +
                "    'table-name' = 'scores'\n" +
                ")");

        tableEnv.executeSql("CREATE TABLE result_table (\n" +
                "    name STRING,\n" +
                "    age INT,\n" +
                "    math_score INT,\n" +
                "    english_score INT\n" +
                ") WITH (\n" +
                "    'connector' = 'print'\n" +
                ")");

        tableEnv.executeSql("INSERT INTO result_table\n" +
                "SELECT s.name, s.age, sc.math_score, sc.english_score\n" +
                "FROM students_table s\n" +
                "JOIN scores_table sc ON s.id = sc.student_id");

    }
}
